package main

import (
	"errors"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"github.com/influxdata/influxdb/uuid"
	jsoniter "github.com/json-iterator/go"
	"log"
	"net/http"
	"os"
	"time"
)

var (
	upGrader = websocket.Upgrader{
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}

	// MessageHandler 回调消息方法
	MessageHandler = func(channel chan []byte) mqtt.MessageHandler {
		return func(client mqtt.Client, msg mqtt.Message) {
			channel <- msg.Payload()
			//fmt.Printf("[TOPIC] %s\n", msg.Topic())
			//fmt.Printf("[MSG] %s\n", msg.Payload())
		}
	}
)

// WsData 消息内容
type EMQonlineData struct {
	Status     int
	UpdateTime int
}
type data struct {
	Token      string `json:"token"`
	UserId     int    `json:"user_id"`
	CustomerId int    `json:"customer_id"`
	Type       string
}

type OnlineData struct {
	Status int
	Error  error
	Data   interface{}
}

func main() {
	r := gin.New()
	r.Use(gin.Logger(), gin.Recovery())

	r.GET("/ws", GetGatewayOnlineStatus)

	if err := r.Run(":8000"); err != nil {
		panic(err)
	}
}

// GetClient 获取EMQ连接属性
// @params Addr string 链接地址 tcp://14.29.243.119:1883
// @params ClientID string 连接ID get_gateway_status[随机]
func GetClient(Addr string, ClientID string, messageFun mqtt.MessageHandler) (mqtt.Client, error) {
	mqtt.DEBUG = log.New(os.Stdout, "", 0)
	mqtt.ERROR = log.New(os.Stdout, "", 0)
	opts := mqtt.NewClientOptions().AddBroker(Addr).SetClientID(ClientID)

	opts.SetKeepAlive(30 * time.Second)

	// 设置消息回调处理函数
	opts.SetDefaultPublishHandler(messageFun)

	opts.SetPingTimeout(1 * time.Second)

	Client := mqtt.NewClient(opts)

	if token := Client.Connect(); token.Wait() && token.Error() != nil {
		return nil, token.Error()
		// panic(token.Error())
	}
	return Client, nil
}

// GetGatewayOnlineStatus 获取网关在线状态
func GetGatewayOnlineStatus(c *gin.Context) {
	// 1.将http请求升级为websocket
	conn, er := upGrader.Upgrade(c.Writer, c.Request, nil)
	if er != nil {
		fmt.Println("upGrader.Upgrade", er)
		return
	}

	out := make(chan struct{}, 1)
	defer close(out)

	go func() {
		for {
			if _, _, err := conn.ReadMessage(); err != nil {
				out <- struct{}{}
				return
			}
		}

	}()

	mqtt.ERROR = log.New(os.Stdout, "[error]", 0)
	opts := mqtt.NewClientOptions().AddBroker("tcp://14.29.243.119:1883").SetClientID(uuid.TimeUUID().String())
	opts.SetKeepAlive(60 * time.Second)
	opts.SetPingTimeout(time.Second)
	opts.SetProtocolVersion(4)

	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	//items := []string{"101000040", "101000028", "101000040", "101000030", "101000031"}
	items := []string{"101000040", "101000028"}

	//for {
	//for _, v := range items {
	//	fmt.Println("[item]", v)
	//	if token := client.Subscribe("/Gateway/Status/"+v, 1, MessageHandler(nil)); token.Wait() && token.Error() != nil {
	//		panic(token.Error())
	//	}
	//
	//	//defer client.Unsubscribe("/Gateway/Status/" + v)
	//}

	//time.Sleep(time.Second * 2)
	//}

	channel := make(chan []byte, 1)

	for _, v := range items {
		if token := client.Subscribe("/Gateway/Status/"+v, 1, MessageHandler(channel)); token.Wait() && token.Error() != nil {
			panic(token.Error())
		}

		defer client.Unsubscribe("/Gateway/Status/" + v)
	}

	fmt.Println("[websocket] receiver")
	// 11.开启无限循环读取到EMQ的返回消息则推送给前端
	for {
		var returnData OnlineData
		var value *EMQonlineData
		cValue := <-channel
		fmt.Println("[cValue]", string(cValue))
		if err := jsoniter.Unmarshal(cValue, &value); err != nil {
			returnData = OnlineData{
				Status: 201,
				Error:  errors.New("解析JSON错误"),
			}
		}
		returnData = OnlineData{
			Status: 200,
			Error:  nil,
			Data: EMQonlineData{
				Status:     value.Status,
				UpdateTime: value.UpdateTime,
			},
		}
		// 12.发送JSON格式的消息
		if ee := conn.WriteJSON(returnData); ee != nil {
			client.Disconnect(250)
			time.Sleep(1 * time.Second)
			fmt.Println("[websocket] out")
			fmt.Println("conn.WriteJSON", ee)
			return
		}

		set := time.Millisecond * 20
		timer := time.NewTimer(set)

		select {
		case <-out:
			fmt.Println("[websocket] out")
			return
		case <-timer.C:
			timer.Reset(set)
		}
		time.Sleep(time.Millisecond * 300)
	}
}
